package cn.kgc.app

import cn.kgc.bean.Ads_log
import cn.kgc.utils.{JdbcUtil, MyKafkaUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 需求:最近一小时某个广告点击量趋势统计
 *
 *   1）开窗确定时间范围；
 *   2）在窗口内将数据转换数据结构为((adid,hm),count);
 *   3）按照广告id进行分组处理，组内按照时分排序。
 *   展示结果:
 *     1：List [15:50->10,15:51->25,15:52->30]
 *     2：List [15:50->10,15:51->25,15:52->30]
 *     3：List [15:50->10,15:51->25,15:52->30]
 */
object LastHourAdCountHandler {
  def main(args: Array[String]): Unit = {

    // 创建时间格式化转换对象  把时间转化为年月日
    val dateformat = new SimpleDateFormat("HH:mm")

    // 创建sparkStreaming 环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("BlackListHandler")
    val ssc = new StreamingContext(conf,Seconds(3))
    // 1）读取Kafka数据之后，并对MySQL中存储的黑名单数据做校验；
    val source = MyKafkaUtil.getKafkaStream("ad", ssc)
    val adsLogDStream = source.map(x => {
      val arr = x.value().split(" ")

      Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4))
    })

    // 1）开窗确定时间范围；
    val windowDStream = adsLogDStream.window(Seconds(12), Seconds(3))
    // 2）在窗口内将数据转换数据结构为((adid,hm),count);
    val transformDStream = windowDStream.transform(rdd => {
      // 得到每条广告每分钟的点击量
      rdd.map(ads_log => {
        ((ads_log.adid, dateformat.format(new Date(ads_log.timestamp))), 1)
      }).reduceByKey(_ + _)
        .map {
          case ((adid, hm), count) => (adid, (hm, count))
        }
        // 根据adid 分组   adid it(hm,count,hm,count,hm,count)
        .groupByKey()
        // 进行组内排序
        .mapValues(v => v.toList.sortBy(_._1))

    })
    transformDStream.print()
    ssc.start()
    ssc.awaitTermination()

  }


}
